package app.dws;

import app.BaseAppV1;
import beans.InterCourseReviewBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DimAsyncFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.AtguiguUtil;
import utils.FlinkSinkUtil;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_0302_DwsInterCourseReviewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_0302_DwsInterCourseReviewWindow().init(
                3302,2,
                "Dws_0302_DwsInterCourseReviewWindow",
                Constant.TOPIC_DWD_DWD_REVIEW_INFO
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 封装到一个 pojo 中
        SingleOutputStreamOperator<InterCourseReviewBean> beanStream = parseToPojo(stream);
        // 2. 开窗聚和
        SingleOutputStreamOperator<InterCourseReviewBean> aggStream = windowAndAgg(beanStream);
        //补充维度
        SingleOutputStreamOperator<InterCourseReviewBean> resultStream = joindim(aggStream);
        // 3. 写出到 clickhouse 中
        writeToClickhouse(resultStream);
    }

    private SingleOutputStreamOperator<InterCourseReviewBean> joindim(SingleOutputStreamOperator<InterCourseReviewBean> aggStream) {
        return AsyncDataStream.unorderedWait(
                aggStream,
                new DimAsyncFunction<InterCourseReviewBean>() {
                    @Override
                    public String table() {
                        return "dim_course_info";
                    }

                    @Override
                    public String id(InterCourseReviewBean input) {
                        return input.getCouseId();
                    }

                    @Override
                    public void addDim(JSONObject dim, InterCourseReviewBean input) {
                        input.setCourseName(dim.getString("course_name"));
                    }
                },
                120, TimeUnit.SECONDS
        );
    }

    private void writeToClickhouse(SingleOutputStreamOperator<InterCourseReviewBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.
                getClickHouseSink("dws_inter_course_review_window", InterCourseReviewBean.class));

    }

    private SingleOutputStreamOperator<InterCourseReviewBean> windowAndAgg(SingleOutputStreamOperator<InterCourseReviewBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <InterCourseReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts)->bean.getTs())
                )
                .keyBy(bean->bean.getCouseId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<InterCourseReviewBean>() {
                            @Override
                            public InterCourseReviewBean reduce(InterCourseReviewBean value1, InterCourseReviewBean value2) throws Exception {
                                value1.setReviewStars(value1.getReviewStars() + value2.getReviewStars());
                                value1.setRc(value1.getRc() + value2.getRc());
                                value1.setGc(value1.getGc() + value2.getGc());

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<InterCourseReviewBean, InterCourseReviewBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<InterCourseReviewBean> elements,
                                                Collector<InterCourseReviewBean> out) throws Exception {
                                InterCourseReviewBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDatTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDatTime(context.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);
                            }
                        }
                );


    }

    private SingleOutputStreamOperator<InterCourseReviewBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, InterCourseReviewBean>() {
                    @Override
                    public InterCourseReviewBean map(JSONObject value) throws Exception {
                        String courseId = value.getString("course_id");
                        Long reviewStars = value.getLong("review_stars");//评价等级
                        long rc = 1L;//用户
                        long gr = 0L;
                        if (reviewStars > 3) {
                            gr=1;//好评用户
                        }
                        Long ts = value.getLong("ts");


                        return new InterCourseReviewBean("","",
                                courseId,"",reviewStars,rc,gr,ts);
                    }
                });

    }
}
